…uster capacity

Best-effort admission check for pipelined groups. All member stages of a group
must run concurrently, so if the group's total task demand exceeds the cluster's
total concurrent-task capacity it can never be co-resident and, lacking an
out-of-band slot reservation, would deadlock (the consumer holds slots waiting
for producer output while the producer cannot get slots to produce). We fail
fast with a clear CONCURRENT_SCHEDULER_INSUFFICIENT_SLOT error instead.

When a pipelined group is about to be co-scheduled, submitStage computes the
group's full pipelined-connected component (pipelinedGroupOf) and compares its
summed task demand against maxConcurrentTasksForStage (production:
sc.maxNumConcurrentTasks for the stage's resource profile). If demand exceeds
capacity, the job is aborted; the already-launched producers are torn down by
the normal job-abort path (cancelRunningIndependentStages).

This is deliberately best-effort, not the atomic gang reservation deferred to a
later hardening step: it compares the whole group's demand against TOTAL
capacity (not free slots), checked once at co-schedule time. That converts the
common under-provisioned case (a group that can never fit) into a clear error
rather than a hang; races against other concurrently admitting work are left to
the future atomic version. Inert for jobs with no pipelined dependency.

maxConcurrentTasksForStage is a protected seam so tests can control reported
capacity without changing the cluster's core count.

Tests (DAGSchedulerSuite): a group too large to co-fit fails fast with
CONCURRENT_SCHEDULER_INSUFFICIENT_SLOT; a group that fits is co-scheduled
normally; producer/consumer task sets marked isPipelined; regular task sets are
not.

Co-authored-by: Isaac

…4.1)

Upgrade the gang-admission slot check from total capacity to currently-FREE
slots: free = per-profile total capacity minus tasks already running for OTHER
work (other groups / regular jobs), excluding the group's own already-running
members. Adds TaskSchedulerImpl.runningTasksForOtherWorkInProfile (single-lock,
resource-profile-scoped, zombie-filtered) and a DAGScheduler seam. Comparing
against total capacity would admit a group that fits in principle but cannot
co-fit beside a busy neighbor, then hang; free-slot admission fails fast instead.
Also bumps the incremental-manager config .version to 4.3.0.

… add a slot-check disable flag

Two updates to the pipelined-group slot admission check (spec S4.1), per an updated spec:

- Count outstanding demand, not just running tasks. The occupancy of OTHER work in a resource
  profile now sums each task set's not-yet-completed tasks (numTasks - tasksSuccessful), i.e.
  running plus enqueued, instead of only the tasks actively running. A neighbor's queued backlog is
  real committed demand that can starve a co-scheduled group, so charging only running tasks could
  admit a group that then hangs once the backlog launches. Renamed TaskSchedulerImpl's helper to
  outstandingTasksForOtherWorkInProfile and the DAGScheduler seam to outstandingTasksForOtherWork
  to reflect the new semantics.

- Add spark.scheduler.pipelinedGroup.slotCheck.enabled (internal, default true). When false,
  pipelinedGroupExceedsCapacity is skipped entirely and a group is co-scheduled unconditionally.
  This mirrors ConcurrentStageDAGScheduler's ability to disable its slot check, for deployments that
  admit capacity out-of-band (e.g. a slot reservation) and own admission themselves.

Tests: the TaskSchedulerImpl unit test now submits more tasks than slots and asserts the count
includes the enqueued tasks; a new DAGScheduler test asserts that with the check disabled an
over-capacity group is co-scheduled rather than failed.

Co-authored-by: Isaac

… the group finishes

Group-observable completion for pipelined groups (spec S5). A stage co-scheduled
with a still-running pipelined producer (a pipelined consumer) must not have its
successful completion processed early: doing so would advance job completion and
cancel the still-running producer (via cancelRunningIndependentStages), or make
the consumer's output observable before the producer's.

handleTaskCompletion: near the top, before any of the event's side effects, if a
Success event belongs to a pipelined consumer with unfinished pipelined
producers, buffer the whole CompletionEvent and return. This defers the entire
event (coarse model), so its side effects (accumulator update, TaskEnd listener
event, stage/job completion) run exactly once -- at replay.

markStageAsFinished: when a pipelined producer finishes, release its deferred
consumers -- but only when the producer's outcome is final. A ShuffleMapStage
that finished without error yet is not isAvailable (an output missing; it will be
resubmitted) is NOT treated as done, so its consumers stay deferred until the
reattempt makes it available. On genuine success the buffered events are
replayed; on producer failure they are dropped (the group reruns, S6) but their
TaskEnd events are still emitted so listeners do not leak active-task accounting.

cleanupStateForJobAndIndependentStages: drop any deferral keyed on a removed
stage and remove it from other consumers' pending-producer sets, so no deferral
outlives its job (e.g. on abort). assertDataStructuresEmpty also checks the
deferral map is empty.

Inert for jobs with no pipelined dependency: the deferral map is never populated,
the completion check is a always-miss map lookup, and release/cleanup are no-ops.

Tests (DAGSchedulerSuite): early-finishing consumer does not end the job or
cancel its running producer; normal producer-then-consumer ordering; buffered
completions dropped when the producer fails; TaskEnd fired exactly once (no
buffer+replay duplication); deferral released only when the producer is
genuinely available.

Co-authored-by: Isaac

…ts, reject dynamic allocation

Two correctness fixes to the pipelined-group slot admission, found in review:

- Skip zombie attempts in outstandingTasksForOtherWorkInProfile. The occupancy count summed
  numTasks - tasksSuccessful over every attempt in taskSetsByStageIdAndAttempt, including zombie
  (superseded) attempts. A retried/killed stage can have both a zombie and a live attempt in the
  map at once, and the live attempt already re-runs the zombie's outstanding tasks -- so counting
  both double-counted that stage's demand, inflating occupancy and potentially failing a pipelined
  group with CONCURRENT_SCHEDULER_INSUFFICIENT_SLOT that would actually fit. Add the !isZombie
  filter used elsewhere on this map.

- Reject a pipelined-shuffle job under dynamic allocation. A pipelined group is gang-scheduled and
  its free-slot check measures currently-active executors; under dynamic allocation a job can be
  submitted before any executor has spun up, so the group would be failed against a transient
  0-slot snapshot even though the cluster would soon have capacity. Barrier scheduling forbids the
  same combination (checkBarrierStageWithDynamicAllocation); pipelined groups now do likewise.
  The former rejectSpeculationWithPipelinedShuffle is generalized to rejectUnsupportedPipelinedJob,
  which rejects both speculation and dynamic allocation (single RDD-graph walk, run only when a
  relevant feature is enabled).

Tests: a zombie + live attempt is counted once, not twice; a pipelined job under dynamic allocation
is rejected while a regular job under dynamic allocation is not.

Co-authored-by: Isaac

…by a later PR in the stack

isPipelinedGroupMember is defined here alongside the other group-topology helpers but is first used
by the group-atomic failure handling added in a later PR of this stack (TaskSet.isPipelined and the
member-failure fail-fast, spec S6). Reviewed in isolation this change reads as introducing an unused
private method; note the forward reference in the scaladoc so it is not mistaken for dead code.

Co-authored-by: Isaac

…unning plus enqueued

The admission check counts OTHER work's OUTSTANDING tasks (running plus
enqueued, `numTasks - tasksSuccessful`) against a resource profile's
capacity -- as `outstandingTasksForOtherWork` / `TaskSchedulerImpl.
outstandingTasksForOtherWorkInProfile` and the config doc already state.
Two spots in DAGScheduler still described this as only the tasks "already
running", which was stale after the count was widened to include enqueued
tasks; the `pipelinedGroupExceedsCapacity` scaladoc even contradicted its
own later "running or enqueued" line. Reword both to "outstanding --
running plus enqueued" so the prose matches the code. Doc/comment only; no
behavior change.

Co-authored-by: Isaac

… admit the group up front

v1 (M1, real-time-mode scope) supports a job that is either all-regular or
all-pipelined, not a mix. This lets an all-pipelined job's whole stage graph be
one pipelined group with no regular prefix, so gang admission can be decided up
front -- before any member stage is submitted -- rather than mid-DAG once a
producer is already running.

Changes in handleJobSubmitted (before createResultStage, so a rejection leaves
no partial scheduler state, exactly like the barrier slot check and the
speculation/DA reject):
 - classifyJobShuffleKinds walks the RDD graph and rejects a job that mixes a
   pipelined shuffle with a regular one, fail-fast.
 - rejectUnadmittablePipelinedGroup computes the whole all-pipelined group's
   concurrent-task demand from the RDD graph and checks it against free slots
   (maxNumConcurrentTasks minus other work's running-plus-enqueued demand, spec
   S4.1); fails the job if it cannot fit. No scheduler retry -- a transient
   shortfall is the caller's to retry (e.g. the streaming batch loop reruns the
   batch).

This is true all-or-nothing gang admission: the whole group is admitted, or the
job fails before any member runs, so a member is never left running while a
sibling cannot get slots. The late slot check in submitStage's co-schedule
branch (which measured a mid-flight snapshot after a producer was already
running) is removed, along with the now-unused pipelinedGroupExceedsCapacity /
pipelinedGroupOf; the capacity/occupancy seams are refactored to be
resource-profile-keyed.

Tests: mixed jobs rejected up front; all-PG group admitted/rejected up front
(2- and 3-stage); the removed mid-DAG prefix tests are dropped as that shape is
no longer supported.

Co-authored-by: Isaac

…oncurrentStageDAGScheduler

Rename the deferred-completion machinery to match the production RTM scheduler
(ConcurrentStageDAGScheduler), so reviewers familiar with that code read the
same concepts here. Pure rename; no behavior change.

 - pipelinedConsumerDeferrals -> dependentStageMap
 - DeferredCompletion         -> DependentStageInfo
 - pendingProducers           -> parents
 - bufferedEvents             -> delayedTaskCompletionEvents

The pipelined-specific helpers (isPipelinedProducer, isPipelinedGroupMember,
submitWaitingPipelinedChildStages, rejectUnadmittablePipelinedGroup) keep their
names -- they belong to the type-driven pipelined model and have no clean
analog in the reference scheduler.

Co-authored-by: Isaac

… check; fix a vacuous admission test

Two test-hygiene fixes from an adversarial review:

- assertDataStructuresEmpty now asserts dependentStageMap.isEmpty, per the
  DAGScheduler class-doc checklist ("when adding a new data structure, update
  DAGSchedulerSuite.assertDataStructuresEmpty ... to catch memory leaks"). The
  new dependentStageMap (the pipelined deferred-completion buffer) was omitted;
  it is always drained today, so this passes -- and now guards a future leak of
  buffered CompletionEvents / a consumer stage left as perpetually-running.

- Retitle/reframe "the group's own running members are not charged against its
  admission": it stubbed the occupancy seam to a constant ignoring
  excludeStageIds, and the up-front admission passes Set.empty (no job stage
  exists yet), so it never exercised member exclusion -- it only asserts a
  demand==free-capacity group is admitted (a real boundary). The member-exclusion
  invariant is genuinely covered in TaskSchedulerImplSuite; the title/comment now
  say what the test actually verifies.

Co-authored-by: Isaac

…reads as unused" and a phantom symbol

The NOTE on isPipelinedGroupMember said it "reads as unused when this change is viewed
alone" and listed `pipelinedGroupOf` as a sibling helper. Both are inaccurate in the
merged view: the method IS used (its call sites -- TaskSet.isPipelined tagging and the
member-FetchFailed group abort -- are part of this stack), and `pipelinedGroupOf` does not
exist. Reword to name the actual call sites and drop the phantom reference.

Co-authored-by: Isaac

…private[scheduler]

The method exists only to back the pipelined-group slot admission check in DAGScheduler; it
does not belong on the public TaskSchedulerImpl surface. Its only callers (DAGScheduler and
TaskSchedulerImplSuite) are in org.apache.spark.scheduler, so private[scheduler] is
sufficient and keeps the new capacity-accounting method out of the public API.

Co-authored-by: Isaac

… the retry behavior

The gang-admission scaladoc likened the check to barrier's slot check. The likeness is only
that both reject before any stage is created and compute demand from the RDD graph; their
retry behavior differs and a reader should not infer equivalence. Spell out the contrast:
barrier re-posts the job and retries its check up to a max-failures bound, whereas pipelined
admission is terminal (one check, then fail) and delegates transient-shortfall retry to the
caller -- scheduler-side PG-admission retry being a post-v1 refinement (spec S4.1).

Co-authored-by: Isaac

…t lands on a torn-down stage

A deferred pipelined-consumer completion fires its TaskEnd inline on first completion; only
the stage/job bookkeeping is re-posted (finishOnly=true) once the producer finishes. The
cancelled-stage guard in handleTaskCompletion (which posts TaskEnd for an event whose stage
is no longer tracked) ran unconditionally and before the finishOnly check. If the group is
torn down (a sibling aborts) between a finishOnly re-post and its dequeue, the replay lands
on the removed stage and the guard would re-post TaskEnd -- double-counting it in listeners
(AppStatusListener's active-task count could go negative). Skip postTaskEnd for a finishOnly
event in that guard: its per-task TaskEnd already fired inline.

A regression test on the failure PR (which has the group-atomic abort this needs) drives a
finishOnly replay onto a torn-down stage and asserts no extra TaskEnd; reverting this guard
makes it fail (TaskEnd count goes up by one).

Co-authored-by: Isaac

…cumulator deltas are not un-merged

The drop-path comment implied full cleanliness ("only the deferred completion bookkeeping
was buffered ... must NOT be applied"). But a deferred consumer's per-task effects, including
updateAccumulators, run inline before buffering, so on a producer-failure drop the already-
applied accumulator deltas are NOT un-merged. Note that this is consistent with base Spark
(accumulators are non-transactional across abort+rerun): the rerun re-delivers under the S5
idempotent-sink contract, and RTM SQL metrics are fresh per batch, so there is no cross-batch
double-count. Comment-only.

Co-authored-by: Isaac

…ments

Strip internal shorthand that a reader of merged Apache Spark cannot resolve
-- version/milestone tags, design-doc spec section numbers, and
fine-grained/coarse model terminology -- from the pipelined-shuffle comments,
one test title, and the mixed-job error message, rewording to plain behavioral
language. Comment, test-title, and error/assertion-string text only; no
behavioral change.

Co-authored-by: Isaac

…fault resource profile

Up-front gang admission measures capacity and occupancy against the default
resource profile, but each stage derives its profile from its RDDs. A pipelined
job whose members use a non-default resource profile could therefore pass
admission against the default profile's free slots and then queue or deadlock in
the profile where its tasks actually run.

Reject such a job up front in rejectUnadmittablePipelinedGroup (before any stage
is created, like the other up-front rejections): walk the group's RDD graph and
fail the job if any member carries a non-default resource profile. The whole
pipelined group is required to run on the default profile -- the assumption the
admission check already relies on. Per-profile admission is a follow-up.

Add a DAGSchedulerSuite test: a pipelined consumer on a custom resource profile
is rejected before any stage is created.

Co-authored-by: Isaac

…hole completion event, remove finishOnly

Group-observable completion (spec S5) held a pipelined consumer that finished ahead of its
still-running producer. The previous "fine-grained" model ran the consumer's per-task side
effects (accumulator update, task-end listener event) inline and deferred only the
stage/job-completion decision; making the side effects run exactly once then required a
CompletionEvent.finishOnly flag so the replayed event would skip the already-applied
per-task half, plus a special guard so a replay landing on a torn-down stage would not
re-post TaskEnd.

Switch to the simpler coarse model, per review feedback on apache#57341: defer the *whole*
CompletionEvent and return before ANY of its side effects run. This makes the side effects
run exactly once by construction -- at replay, when releaseDeferredPipelinedConsumers
re-posts the buffered event and it re-enters handleTaskCompletion normally -- so:

- CompletionEvent.finishOnly is removed entirely.
- handleTaskCompletion has a single deferral check up front (before updateAccumulators /
  postTaskEnd), and the normal side-effect path is unguarded again (identical to base
  Spark), so the torn-down-stage postTaskEnd guard is no longer needed.
- The producer-failure drop path emits the buffered tasks' TaskEnd events (they genuinely
  succeeded, so active-task-tracking listeners must see them finish) but does NOT run the
  stage/job-completion bookkeeping, since the group is rerun.

Trade-off (accepted): a consumer that finishes early has its task-level listener events
(TaskStart already fired; TaskEnd, accumulator updates) held until the producer's fate is
known, rather than flowing in real time. The stage/job-completion decision -- the part that
would wrongly cancel the still-running producer or expose output early -- is deferred either
way; only the per-task event *timing* differs.

Tests: the two deferral tests now assert the coarse timing (no consumer TaskEnd until the
producer finishes; drop path emits TaskEnd but no result; replay still yields exactly 4
TaskEnd events with no buffer+replay duplication). Reverting the change (emitting TaskEnd
inline before the deferral buffer) makes them fail (TaskEnd count 2 instead of 0 while the
producer runs). Full DAGSchedulerSuite green (344 tests, both variants).

Co-authored-by: Isaac

…own, not just on the release path

The coarse deferral model buffers a pipelined consumer's whole completion event -- including its
TaskEnd -- until group completion. releaseDeferredPipelinedConsumers flushes those buffered
TaskEnds on both its replay (producer succeeded) and drop (producer failed) paths, and it is
normally reached for every producer before job cleanup runs (cancelRunningIndependentStages
finishes each running producer first).

But cleanupStateForJobAndIndependentStages -> removeStage discarded a consumer's deferral with a
bare `dependentStageMap -= stage`, without flushing its buffered TaskEnds. That silently loses
them in a case the release path does not cover: a producer left in resubmit limbo -- finished
with no error yet not available, so producerAboutToResubmit retained the deferral AND
markStageAsFinished removed the producer from runningStages -- is skipped by
cancelRunningIndependentStages (it is neither running nor failed). If the job is then cancelled
or aborted, cleanup drops the consumer's buffered completion without ever emitting its TaskEnds,
so a listener that tracks active tasks (e.g. AppStatusListener) leaks them as perpetually
running. This was impossible before the coarse-model change, when TaskEnd fired inline on first
completion and nothing was buffered to lose.

Fix: at removeStage, drain the removed consumer's buffered delayedTaskCompletionEvents through
postTaskEnd (as the producer-failed drop path does) before discarding the deferral. The buffered
successes are still not applied as results -- this is a job failure / cancellation -- only their
TaskEnds are flushed.

Test: a regression test drives a producer into resubmit limbo, cancels the job, and asserts the
2 buffered consumer TaskEnds are flushed (and no result applied). Reverting the fix makes it fail
(0 TaskEnds instead of 2). Full DAGSchedulerSuite green.

Co-authored-by: Isaac

Two documentation-only corrections from review (cloud-fan), no behavior change:

- TaskSchedulerImpl.outstandingTasksForOtherWorkInProfile scaladoc: "the slots free by everything
  else" reversed the relationship (other work consumes slots, it does not free them). Reword to
  "the slots left free after accounting for everything else in the SAME resource profile".

- DAGScheduler.handleMapStageSubmitted comment: the parenthetical said the result-job path
  (handleJobSubmitted) "only rejects the speculation case". It rejects both speculation and dynamic
  allocation via rejectUnsupportedPipelinedJob; correct the summary.

Co-authored-by: Isaac

…test assert

The assert message in "job teardown flushes a buffered consumer's TaskEnds" was
101 chars, over the 100-char scalastyle limit for test sources. Trim it.

Co-authored-by: Isaac

… per-task side effects

Address review (cloud-fan): make the group-observable-completion comment explicit that
buffering the whole CompletionEvent DEFERS all three side effects -- the accumulator update,
the task-end listener event, and stage/job completion -- not just the stage/job-completion
bookkeeping. Also state the listener-visible consequence: a consumer task that finishes ahead
of its producer emits no SparkListenerTaskEnd until the producer finishes, so the Spark UI
reports that already-succeeded task as still running for the producer's remaining lifetime.

Comment-only; no behavior change.

Co-authored-by: Isaac

…producer fails

releaseDeferredPipelinedConsumers decided drop-vs-replay only when a consumer's last pending
producer finished, keyed on that producer's outcome. For a consumer co-scheduled with more than
one pipelined producer (fan-in), producer A could fail while B was still pending -- A was removed
from the pending set but no drop occurred, and A's failure was forgotten -- so when B later
succeeded it emptied the set on the replay branch and the consumer's buffered successes were
replayed even though they were computed against A's now-invalid output.

Fix: when a producer fails, drop the consumer's deferral immediately -- flush its buffered
TaskEnds and remove the entry -- regardless of whether it still has other pending producers. A
pipelined group is failed as a unit (S6), so once any producer fails the buffered successes must
not be applied. Dropping now (rather than remembering the failure until the last producer
finishes) also leaves no deferral state that could outlive the call and be reused by a later
re-co-scheduling of the same stage. A surviving producer that finishes afterward simply finds the
consumer already gone (a clean no-op). Producer success is unchanged: remove the finished producer
and replay only once the last one has succeeded.

Recovery from a drop, if any, is a caller rerun of the whole group (a new job, S6); the scheduler
does not rerun the group in place.

Test: a two-producer (fan-in) case where A fails and B then succeeds, asserting the consumer's
deferral is dropped as soon as A fails and B's later success does not replay it. Reverting to the
old parents-empty gating makes it fail (the successes replay). Full DAGSchedulerSuite green.

Co-authored-by: Isaac

… drop path

The immediate-drop on producer failure treats producerFailed as terminal. That holds for every
producer-failure path that reaches releaseDeferredPipelinedConsumers except one: a barrier stage
that fails a task and is then resubmitted (its handler calls markStageAsFinished(errorMessage)
without willRetry=true, then resubmits). For a fan-in consumer, dropping its deferral for a barrier
producer that will retry loses the consumer's buffered results.

A barrier stage in a pipelined group is rejected up front by a later PR in this stack
(checkPipelinedProducerSupported), which removes this path entirely; on this PR standalone it is a
known, narrow limitation (needs a barrier producer feeding a fan-in consumer). Document it inline
rather than adding the rejection early.

Comment-only; no behavior change.

Co-authored-by: Isaac

…d consumer" to the stage

Address review nit (cloud-fan): the parenthetical (a "pipelined consumer") sat after "a pipelined
producer that is still running", so it read as though the producer is the consumer. Move it to
follow "stage" -- the co-scheduled stage is the consumer.

Comment-only; no behavior change.

Co-authored-by: Isaac